package com.qch.kafka.demo.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author qch
 * @Date 2021/6/6
 */
public class MyTransactionalProducer {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.142.135:9092");
        configs.put("key.serializer",StringSerializer.class);
        configs.put("value.serializer",StringSerializer.class);
        configs.put(ProducerConfig.CLIENT_ID_CONFIG,"tx_producer");
        //使用事务就必须设置
        configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my_tx_id");
        configs.put(ProducerConfig.ACKS_CONFIG, "all");
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);
        producer.initTransactions();
        try {
            producer.beginTransaction();
            producer.send(new ProducerRecord<>("tp_tx_01","txkey1","tx_msg_4"));
            producer.send(new ProducerRecord<>("tp_tx_01","txkey1","tx_msg_5"));
            producer.send(new ProducerRecord<>("tp_tx_01","txkey1","tx_msg_6"));
            int i=1/0;
            producer.commitTransaction();
        }catch (Exception e){
            e.printStackTrace();
            //事务回滚
            producer.abortTransaction();
        }finally {
            producer.close();
        }
    }
}
